package com.disruptor.demo;

import org.jctools.queues.SpscArrayQueue;

import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicLong;

/**
 * 基于JCTools和FileChannel的高性能文件存储线程
 */
public class JCToolsFileStoreThread extends Thread {

    public final SpscArrayQueue<Long> queue;
    private final AtomicBoolean running = new AtomicBoolean(true);
    private final long times;
    private final long l;

    public JCToolsFileStoreThread(int queueSize,  long times,long l) {
        this.queue = new SpscArrayQueue<>(queueSize);
        this.times=times;
        this.l=l;
    }


    public boolean offerData(Long data) {
        return queue.offer(data);
    }

    public void close() {
        running.set(false);
    }

    @Override
    public void run() {
        try {
            AtomicLong i= new AtomicLong(1);
            while (running.get()|| !queue.isEmpty()) {
                // 批量消费队列

                int drained = queue.drain(body -> {
                    if(i.longValue()==times){
                        System.out.println(i.intValue());
                        System.out.println(System.currentTimeMillis()-l);
                    }
                    i.getAndIncrement();
                }, 10);

                // 如果本轮没有处理任何数据，且仍在运行，则短暂休眠
                if (drained == 0 && running.get()) {
                    Thread.yield();
                }
            }

        } catch (Exception e) {
            throw new RuntimeException("文件写入线程异常", e);
        } finally {

        }
    }
}